package com.bw.util;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;


public abstract class MyDIMRequestIO<IN, OUT> extends RichAsyncFunction<IN, OUT> {

    private Connection connection;
    private ThreadPoolExecutor threadPoolExecutor;

    public static final String PHOENIX_DRIVER = "org.apache.phoenix.jdbc.PhoenixDriver";

    public static final String PHOENIX_SERVER = "jdbc:phoenix:hadoop101:2181";

    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName(PHOENIX_DRIVER);
        connection = DriverManager.getConnection(PHOENIX_SERVER);

        threadPoolExecutor = ThreadPoolUtil.getThreadPool();
    }


    @Override
    public void asyncInvoke(IN in, final ResultFuture<OUT> resultFuture) throws Exception {

        threadPoolExecutor.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    Object key = selectKey(in);
                      if (key == null || "".equals(key)){
                        key = "1";
                      }
                    String replace = buildSql().replace("?", key.toString());
                    List<JSONObject> selectXxFromXxx = DimUtil.<JSONObject>queryList(connection, replace, JSONObject.class, true);
                    if (selectXxFromXxx == null || selectXxFromXxx.size() == 0){
                        System.out.println("ERROR : sql key is null : " + replace);
                    }
                    OUT join = join(in, selectXxFromXxx.get(0));

                    resultFuture.complete(Collections.singleton(join));

                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }

    public abstract Object selectKey(IN in);

    public abstract String buildSql();

    public abstract OUT join(IN in, JSONObject jsonObject);

    @Override
    public void timeout(IN input, ResultFuture<OUT> resultFuture) throws Exception {
        super.timeout(input, resultFuture);
    }
}

class ThreadPoolUtil {

    private static ThreadPoolExecutor threadPoolExecutor = null;

    private ThreadPoolUtil() {
    }

    public static ThreadPoolExecutor getThreadPool() {

        if (threadPoolExecutor == null) {
            synchronized (ThreadPoolUtil.class) {
                if (threadPoolExecutor == null) {
                    threadPoolExecutor = new ThreadPoolExecutor(8,
                            16,
                            1L,
                            TimeUnit.MINUTES,
                            new LinkedBlockingDeque<>());
                }
            }
        }

        return threadPoolExecutor;
    }
}